package com.sub.spark.core.rdd.extend;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;

import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

/**
 * RDD依赖关系
 *
 * 窄依赖表示每一个父RDD的Partition最多被子RDD的一个Partition使用（一对一or多对一），窄依赖我们形象的比喻为独生子女。
 * 宽依赖表示同一个父RDD的Partition被多个子RDD的Partition使用（一对多or多对多），宽依赖我们形象的比喻为超生。
 * 具有宽依赖的transformations包括：sort、reduceByKey、groupByKey、join和调用rePartition函数的任何操作。
 * 在不影响业务要求的情况下，要尽量避免使用有宽依赖的转换算子，因为有宽依赖，就一定会走shuffle，影响性能。
 *      * 扩展：
 *      * 1、Stage任务划分
 *      * JOB --> Stage --> Task
 *      * 2、DAG图
 *      * DAG（Directed Acyclic Graph）有向无环图是由点和线组成的拓扑图形，
 *      * 该图形具有方向，不会闭环。例如，DAG记录了RDD的转换过程和任务的阶段。
 *      * *********************
 *      * RDD任务切分中间分为：Application、Job、Stage和Task
 *      * （1）Application：初始化一个SparkContext即生成一个Application；
 *      * （2）Job：一个Action算子就会生成一个Job；
 *      * （3）Stage：Stage等于宽依赖的个数加1；
 *      * （4）Task：一个Stage阶段中，最后一个RDD的分区个数就是Task的个数。
 *      * 注意：Application->Job->Stage->Task每一层都是1对n的关系。
 *
 */
public class RDDLineage {
    public static void main(String[] args) {

        // 1.创建配置对象
        SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("sparkCore");

        // 2. 创建sparkContext
        JavaSparkContext sc = new JavaSparkContext(conf);

        // 3. 编写代码
        JavaRDD<String> lineRDD = sc.textFile("input/2.txt");
        System.out.println(lineRDD.toDebugString());
        System.out.println("-------------------");

        JavaRDD<String> wordRDD = lineRDD.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public Iterator<String> call(String s) throws Exception {
                List<String> stringList = Arrays.asList(s.split(" "));
                return stringList.iterator();
            }
        });
        System.out.println(wordRDD.toDebugString());
        System.out.println("-------------------");

        JavaPairRDD<String, Integer> tupleRDD = wordRDD.mapToPair(new PairFunction<String, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(String s) throws Exception {
                return new Tuple2<>(s, 1);
            }
        });
        System.out.println(tupleRDD.toDebugString());
        System.out.println("-------------------");

        JavaPairRDD<String, Integer> wordCountRDD = tupleRDD.reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer v1, Integer v2) throws Exception {
                return v1 + v2;
            }
        });
        System.out.println(wordCountRDD.toDebugString());
        System.out.println("-------------------");
        // 4. 关闭sc
        sc.stop();
    }
}